package com.peak.code;

import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Arrays;
import java.util.List;

/**
 * @date 2021/10/31 2:48 PM
 * @author likai
 * @email likai9376@163.com
 * @desc 
 */
public class Demo01 {


    public static void main(String[] args) {
//        testJust();
//        testArrays();
//        testRange();

        testSubject();
    }



    public static void testJust(){
        Mono<String> hello = Mono.just("hello mono");
        hello.subscribe(System.out::println);


        Flux<String> hello01 = Flux.just("hello flux");
        hello01.subscribe(System.out::println);
    }


    public static void testArrays(){
        List<String> iterable = Arrays.asList("foo", "bar", "foobar");
        Flux<String> stringFlux = Flux.fromIterable(iterable);

        stringFlux.subscribe(System.out::println);
    }


    public static void testRange(){
        Flux<Integer> range = Flux.range(5, 3);
        range.subscribe(System.out::println);
    }




    public static void testSubject(){
        Flux<Integer> ints = Flux.range(1, 4)
                .map(i -> {
                    if (i <= 6) return i;
                    throw new RuntimeException("Got to 4");
                });
        Disposable done = ints.subscribe(i -> System.out.println(i),
                error -> System.err.println("Error: " + error),
                () -> System.out.println("done"),
                sub -> sub.request(3));//限制上游请求的数量



    }


}
